package com.sdses.flink.function;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.ArrayList;

public class FlinkSqlJsonArrayExample {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 注册自定义函数（这里注册上面定义的ConcatFunction）
        tableEnv.createTemporaryFunction("concat_func", ConcatFunction.class);

        // 3. 创建数据源表（这里以内存中的数据集作为示例，模拟两列数据：id和name）
        String sourceDDL = "CREATE TABLE source_table (\n" +
                "    id INT,\n" +
                "    name STRING\n" +
                ") WITH (\n" +
                "    'connector' = 'values',\n" +
                "    'data' = '[(1, \"Alice\"),(2, \"Bob\")]'\n" +
                ")";
        tableEnv.executeSql(sourceDDL);

        // 4. 使用 Flink SQL 语句查询数据，并构建 JSON 数组（利用内置函数和自定义函数等）
        Table resultTable = tableEnv.sqlQuery("" +
                "SELECT JSON_ARRAYAGG(" +
                "       JSON_OBJECT('id', id, 'name', name)" +
                "   ) AS json_array_result " +
                "FROM source_table");

        // 5. 将查询结果转换为 DataStream 以便进一步处理
        DataStream<String> resultStream = tableEnv.toDataStream(resultTable, String.class);

        // 6. 收集并处理结果，将其转换为最终的 JSON 数组字符串
        List<String> jsonArrayList = new ArrayList<>();
        resultStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                jsonArrayList.add(value);
            }
        });

        // 将收集到的结果（这里假设只有一条记录，因为我们是汇总成一个 JSON 数组）转换为最终的 JSON 数组字符串格式
        ObjectMapper objectMapper = new ObjectMapper();
        String finalJsonArray = objectMapper.writeValueAsString(jsonArrayList.get(0));
        System.out.println("最终的JSON数组: " + finalJsonArray);

        // 7. 执行 Flink 任务
        env.execute("Flink SQL Json Array Example");
    }

    // 自定义一个简单的字符串拼接函数

    public static class ConcatFunction extends ScalarFunction {
        public String eval(String str1, String str2) {
            return str1 + str2;
        }
    }
}